package com.zcccccy.rocketmq.producer.service;

import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @projectName: zcccccy
 * @packageName: com.zcccccy.extend.rocketmq.producer.service
 * @author: zcy
 * @createTime: 2024/1/29
 * @productName: For Mac IntelliJ IDEA
 */
@Service
public class MessageProducer {

    // TODO 注意：RocketMQ5.0在创建Topic时会指定消息类型，默认为普通消息。
    // TODO 注意：RocketMQ5.0不允许推送不同与Topic消息类型的消息到Topic

    // TODO 本项目只对接了普通消息和定时/延时消息
    // 普通消息 ： https://rocketmq.apache.org/zh/docs/featureBehavior/01normalmessage
    // 定时/延时消息：https://rocketmq.apache.org/zh/docs/featureBehavior/02delaymessage
    // 顺序消息：https://rocketmq.apache.org/zh/docs/featureBehavior/03fifomessage
    // 事务消息：https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage

    // TODO 目前只有同步消息，异步消息后续有时间写

    private static final Logger log = LoggerFactory.getLogger(MessageProducer.class);

    @Autowired
    private Producer producer;

    /**
     * 发送普通消息
     *
     * @param topic
     *            主题
     * @param body
     *            消息体
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 19:22
     **/
    public String sendMessage(String topic, String body) {

        return sendMessage(topic, null, body);
    }

    /**
     * 发送普通消息
     *
     * @param topic
     *            主题
     * @param messageKey
     *            key
     * @param body
     *            消息体
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 19:22
     **/
    public String sendMessage(String topic, String messageKey, String body) {

        return sendMessage(topic, "tag", messageKey, body);
    }

    /**
     * 发送普通消息
     * 
     * @param topic
     *            主题
     * @param tag
     *            tag
     * @param messageKey
     *            key
     * @param body
     *            消息体
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendMessage(String topic, String tag, String messageKey, String body) {

        log.info("发送消息：【主题】：{}, 【tag】：{},【消息key】：{}【消息体】：{}", topic, tag, messageKey, body);

        // 定时/延时消息发送
        MessageBuilder messageBuilder = new MessageBuilderImpl();
        Message message = messageBuilder.setTopic(topic)
            // 设置消息索引键，可根据关键字精确查找某条消息。
            .setKeys(messageKey)
            // 设置消息Tag，用于消费端根据指定Tag过滤消息。
            .setTag(tag)
            // 消息体
            .setBody(body.getBytes()).build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);

            log.info("消息发送成功：【消息id】：{}", sendReceipt.getMessageId());

            return sendReceipt.getMessageId().toString();
        } catch (ClientException e) {
            log.info("消息发送失败");
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 发送延时消息
     *
     * @param topic
     *            主题 key
     * @param body
     *            消息体
     * @param duration
     *            延时时间/毫秒
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendDelayMessage(String topic, String body, Long duration) {
        return sendDelayMessage(topic, null, body, duration);
    }

    /**
     * 发送延时消息
     *
     * @param topic
     *            主题
     * @param messageKey
     *            key
     * @param body
     *            消息体
     * @param duration
     *            延时时间/毫秒
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendDelayMessage(String topic, String messageKey, String body, Long duration) {
        return sendDelayMessage(topic, "tag", messageKey, body, duration);
    }

    /**
     * 发送延时消息
     *
     * @param topic
     *            主题
     * @param tag
     *            tag
     * @param messageKey
     *            key
     * @param body
     *            消息体
     * @param duration
     *            延时时间/毫秒
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendDelayMessage(String topic, String tag, String messageKey, String body, Long duration) {

        // 延迟后的时间戳。
        Long deliverTimeStamp = System.currentTimeMillis() + duration;

        return sendTimedMessage(topic, tag, messageKey, body, deliverTimeStamp);
    }

    /**
     * 发送定时消息
     *
     * @param topic
     *            主题 key
     * @param body
     *            消息体
     * @param deliverTimeStamp
     *            时间戳
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendTimedMessage(String topic, String body, Long deliverTimeStamp) {

        return sendTimedMessage(topic, null, body, deliverTimeStamp);
    }

    /**
     * 发送定时消息
     *
     * @param topic
     *            主题
     * @param messageKey
     *            key
     * @param body
     *            消息体
     * @param deliverTimeStamp
     *            时间戳
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendTimedMessage(String topic, String messageKey, String body, Long deliverTimeStamp) {

        return sendTimedMessage(topic, "tag", messageKey, body, deliverTimeStamp);
    }

    /**
     * 发送定时消息
     *
     * @param topic
     *            主题
     * @param tag
     *            tag
     * @param messageKey
     *            key
     * @param body
     *            消息体
     * @param deliverTimeStamp
     *            时间戳
     * @return java.lang.String
     * @author zcy
     * @createTime 2024/1/29 17:31
     **/
    public String sendTimedMessage(String topic, String tag, String messageKey, String body, Long deliverTimeStamp) {

        log.info("发送消息：【主题】：{}, 【tag】：{},【消息key】：{}【消息体】：{}【时间戳】：{}", topic, tag, messageKey, body, deliverTimeStamp);

        // 定时/延时消息发送
        MessageBuilder messageBuilder = new MessageBuilderImpl();
        Message message = messageBuilder.setTopic(topic)
            // 设置消息索引键，可根据关键字精确查找某条消息。
            .setKeys(messageKey)
            // 设置消息Tag，用于消费端根据指定Tag过滤消息。
            .setTag(tag)
            // 时间戳
            .setDeliveryTimestamp(deliverTimeStamp)
            // 消息体
            .setBody(body.getBytes()).build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);

            log.info("消息发送成功：【消息id】：{}", sendReceipt.getMessageId());

            return sendReceipt.getMessageId().toString();
        } catch (ClientException e) {
            log.info("消息发送失败");
            e.printStackTrace();
        }

        return null;
    }

}
